<?php
include(__DIR__ . '/config.php');
use PhpAmqpLib\Wire\AMQPTable;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Connection\AMQPStreamConnection;

/**
 * 死信队列测试
 * 1、创建两个交换器 exchange.normal 和 exchange.dlx, 分别绑定两个队列 queue.normal 和 queue.dlx
 * 2、把 queue.normal 队列里面的消息配置过期时间，然后通过 x-dead-letter-exchange 指定死信交换器为 exchange.dlx
 * 3、发送消息到 queue.normal 中，消息过期之后流入 exchange.dlx，然后路由到 queue.dlx 队列中，进行消费
 */

// todo 更改配置
$connection = new AMQPStreamConnection(HOST, PORT, USER, PASS, VHOST);

$channel = $connection->channel();

$channel->exchange_declare('exchange.dlx', AMQPExchangeType::DIRECT, false, true);

$channel->exchange_declare('exchange.normal', AMQPExchangeType::FANOUT, false, true);

$args = new AMQPTable();

// 消息过期方式：设置 queue.normal 队列中的消息10s之后过期
$args->set('x-message-ttl', 10000);

// 设置队列最大长度方式： x-max-length
//$args->set('x-max-length', 1);

//给队列queue.normal绑定死信队列
//设置死信队列绑定的交换机
$args->set('x-dead-letter-exchange', 'exchange.dlx');

//绑定死信队列的路右键
$args->set('x-dead-letter-routing-key', 'routingkey');

$channel->queue_declare('queue.normal', false, true, false, false, false, $args);
$channel->queue_declare('queue.dlx', false, true, false, false);

$channel->queue_bind('queue.normal', 'exchange.normal');
$channel->queue_bind('queue.dlx', 'exchange.dlx', 'routingkey');

//开启监听模式
$channel->confirm_select();

//监听数据,成功
$channel->set_ack_handler(function (AMQPMessage $message){
    //update 订单表 set is_send_succ=ture
    echo "Message acked with content " . $message->body . PHP_EOL;
});

//监听数据,失败
$channel->set_nack_handler(function (AMQPMessage $message){
    //update 订单表 set is_send_succ=false
    echo "Message nacked with content " . $message->body . PHP_EOL;
});

$i = 1;
while ($i <= 100){
    $content = '这是消息内容' . $i++;
    //实力消息类
    // 设置delivery_mode为DELIVERY_MODE_PERSISTENT，表示将消息持久化
    $msg = new AMQPMessage($content,['content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);

    $exchange_name = 'exchange.normal';

    $channel->basic_publish($msg,$exchange_name,'rk');
}

//监听写入
$channel->wait_for_pending_acks();

$channel->close();
$connection->close();